package com.demo.api.dataflow;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.demo.api.model.Order;

import java.time.LocalTime;
import java.util.ArrayList;
import java.util.List;

import static java.util.stream.Collectors.toList;

/**
 * 处理流式作业，分为数据抓取（fetchData）和数据处理（processData）
 * 流程：先抓取数据-》处理数据-》再次抓取数据-》如果数据存在则继续处理，如果数据不存在则任务结束-》等待任务定时规则，下次触发
 * 应用场景：适用于不间歇的数据处理，一直产生数据一直处理，比如第三方订单的抓取
 *
 * 实现DataflowJob接口，注意DataflowJob<T>泛型规定了抓取数据的返回类型，和处理数据的传入参数类型一致
 *
 */
public class MyDataflowJob implements DataflowJob<Order> {


  private List<Order> orders = new ArrayList<>();

  //初始化数据，100条未处理数据，每次抓取10条，更新数据状态为已处理
  {
    for (int i = 1; i <= 100; i++) {
      Order order = new Order();
      order.setOrderId(i);
      //未处理
      order.setStatus(0);
      orders.add(order);
    }
  }


  @Override
  public List<Order> fetchData(ShardingContext shardingContext) {
    //抓取数据规则：订单号 % 分片总数 == 当前分片项
    List<Order> orderList = orders.stream().filter(o -> o.getStatus() == 0)
        .filter(o -> o.getOrderId() % shardingContext.getShardingTotalCount() == shardingContext.getShardingItem())
        .collect(toList());

    List<Order> subList = null;
    if (orderList != null && orderList.size() > 0) {
      // 每次只取10条
      subList = orderList.subList(0, 10);
    }
    // 模拟抓取数据耗时
    try {
      Thread.sleep(3000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    LocalTime time = LocalTime.now();
    System.out.println(time + ",我是分片项：" + shardingContext.getShardingItem() + ",我抓取的数据是：" + subList);
    return subList;
  }

  @Override
  public void processData(ShardingContext shardingContext, List<Order> data) {
    data.forEach(o -> o.setStatus(1));
    // 模拟处理数据耗时
    try {
      Thread.sleep(5000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    LocalTime time = LocalTime.now();
    System.out.println(time + ",我是分片项：" + shardingContext.getShardingItem() + ",我正在处理数据！");
  }
}
